package com.cskaoyan.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @program: csmall-practice
 * @description:
 * @author: Lexis
 * @create: 2022-05-29 15:30
 **/
@Slf4j
@Component
public class PromoOrderProducer {

    TransactionMQProducer producer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    RedissonClient redissonClient;

    @Autowired
    CacheManager cacheManager;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 秒杀服务的本地事物  ——> 扣减秒杀商品库存
                Map<String, Long> param = (Map<String, Long>) arg;
                Long psId = param.get("psId");
                Long productId = param.get("productId");

                String localTransactionKey = "promo_local_transaction:" + msg.getTransactionId();
                Integer effectiveRow = 0;

                // 获取锁
                String lockKey = "promo_item_stock:" + psId + ":" + productId;
                RLock lock = redissonClient.getLock(lockKey);

                // 加锁
                // 某一个线程在扣减库存之前，先获取分布式锁，并加锁
                //  如果一个线程加锁失败，会因为加锁失败而阻塞等待
                lock.lock();
                try {
                    effectiveRow = promoItemMapper.decreaseStock(productId, psId);
                    log.info("库存扣减成功！");
                } finally {
                    lock.unlock();
                }


                if (effectiveRow < 1) {
                    // 秒杀商品售罄, 不需要让订单服务生成秒杀商品订单

                    cacheManager.setCache(localTransactionKey, "rollback", 1);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                // 秒杀商品扣减库存成功(本地事物执行成功)
                cacheManager.setCache(localTransactionKey, "commit", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String localTransactionKey = "promo_local_transaction:" + msg.getTransactionId();
                // 获取本地事物的执行结果
                String result = cacheManager.checkCache(localTransactionKey);
                if (result == null || result.trim().isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                }

                if ("rollback".equals(result)) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }


                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    // 方法的返回值，表示订单服务是否能消费到事务消息
    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest request, Long psId, Long productId) {
        String requestJson = JSON.toJSONString(request);

        // 放本地事务执行所需要的参数
        Map<String, Long> paramMap = new HashMap<>();
        paramMap.put("psId", psId);
        paramMap.put("productId", productId);

        // 将数据封装到事务消息中去
        Message message = new Message();
        message.setTopic("promo_order");
        message.setBody(requestJson.getBytes(StandardCharsets.UTF_8));

        TransactionSendResult sendResult = null;
        try {
            sendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            // 事务消息发送成功，且一定会被订单服务消费到
            log.info("事务消息发送成功");
            return true;
        }
        // 事务消息发送失败 或 事务消息发送成功，但本地事务执行失败
        return false;
    }


}
